package main

import (
	"fmt"
	"testing"
)

func main()  {
	//一些便利的生成器
	//pipeline的生成器是将一组离散值转换为channel撒谎那个的值流的任何函数

	//repeat的生成器
	repeat := func(
		done <-chan interface{},
		values ...interface{},
	)<-chan interface{} {
		valueStream := make(chan interface{})
		go func() {
			defer close(valueStream)
			for {
				for _,v := range values {
					select {
					case <-done:
						return
					case valueStream <- v:
					}
				}
			}
		}()
		return valueStream
	}

	//通用pipeline stage,重复使用时很有用
	take := func(
		done <-chan interface{},
		valueStream <-chan interface{},
		num int,
	) <-chan interface{} {
		takeStream := make(chan interface{})
		go func() {
			defer close(takeStream)
			for i:=0;i<num;i++ {
				select {
				case <-done:
					return
				case takeStream <- valueStream:
				}
			}
		}()
		return takeStream
	}

	done := make(chan interface{})
	defer close(done)

	for num := range take(done,repeat(done,1),10) {
		fmt.Printf("%v",num)
	}



	//创建一个重复调用函数的生成器repeatFn
	//通用pipeline stage,重复使用时很有用
	take := func(
		done <-chan interface{},
		valueStream <-chan interface{},
		num int,
	) <-chan interface{} {
		takeStream := make(chan interface{})
		go func() {
			defer close(takeStream)
			for i:=0;i<num;i++ {
				select {
				case <-done:
					return
				case takeStream <- valueStream:
				}
			}
		}()
		return takeStream
	}


	//创建一个重复调用函数的生成器repeatFn
	repeatFn := func(
		done <-chan interface{},
		fn func() interface{},
	) <-chan interface{} {
		valueStream := make(chan interface{})
		go func() {
			defer close(valueStream)
			for {
				select {
				case <-done:
					return
				case valueStream <- fn():
				}
			}
		}()
		return valueStream
	}

	//生成10个随机数字
	done := make(chan interface{})
	defer close(done)

	rand := func() interface{} { return rand.Int()}    //导入库"math/rand"
	for num := range take(done,repeatFn(done,rand),10) {
		fmt.Println(num)
	}


	//toString pipeline stage
	//报错
	toString := func(
		done <-chan interface{},
	) <-chan string {
		stringStream := make(chan string)
		go func() {
			defer close(stringStream)
			for v := range valueStream {
				select {
				case <-done:
					return
				case stringStream <- v.(string):
				}
			}
		}()
		return stringStream
	}

	done := make(chan interface{})
	defer close(done)

	var message string
	for token := range toString(done,take(done,repeat(done,"I","am."),5)) {
		message += token
	}
	fmt.Pringf("message: %s...",message)


	//基准测试函数，一个测试通用stage，另一个测试特定stage
	func BenchmarkGeneric(b *testing.B) {
		done := make(chan interface{})
		defer close(done)

		b.ResetTimer()
		for range toString(done,take(done,repeat(done,"a"),b.N)) {
		}
	}

	func BenchmarkTyped (b *testing.B) {
		repeat := func(done <-chan interface{},values ...string) <-chan string {
			valueStream := make(chan string)
			go func() {
				defer close(valueStream)
				for {
					for _,v := range values {
						select {
						case <-done:
							return
						case valueStream <- v:
						}
					}
				}
			}()
			return valueStream
		}

		take := func(
			done <-chan interface{},
			valueStream <-chan string,
			num int,
		) <-chan string {
			takeStream := make(chan string)
			go func() {
				defer close(takeStream)
				for i := num; i>0 || i == -1; {
					if i != -1 {
						I --
					}
					select {
					case <-done:
						return
					case takeStream <- <-valueStream:
					}
				}
			}()
			return takeStream
		}
		done := make(chan interface{})
		defer close(done)

		b.ResetTimer()
		for range take(done,repeat(done,"a"),b.N){
		}
	}



}



